package com.wy.kafka.consumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

/**
 * 利用线程池解决多线程问题,只能处理不太严谨的数据,更快,消耗更小
 *
 * @author 飞花梦影
 * @date 2022-07-25 23:42:28
 * @git {@link https://gitee.com/dreamFlyingFlower}
 */
public class MyConsumerThreadPool {

	private final static String TOPIC_NAME = "dream-topic";

	public static void main(String[] args) throws InterruptedException {
		String brokerList = "192.168.1.150:9092";
		String groupId = "test";
		int workerNum = 5;

		CunsumerExecutor consumers = new CunsumerExecutor(brokerList, groupId, TOPIC_NAME);
		consumers.execute(workerNum);

		Thread.sleep(1000000);

		consumers.shutdown();
	}

	// Consumer处理
	public static class CunsumerExecutor {

		private final KafkaConsumer<String, String> consumer;

		private ExecutorService executors;

		public CunsumerExecutor(String brokerList, String groupId, String topic) {
			Properties props = new Properties();
			props.put("bootstrap.servers", brokerList);
			props.put("group.id", groupId);
			props.put("enable.auto.commit", "true");
			props.put("auto.commit.interval.ms", "1000");
			props.put("session.timeout.ms", "30000");
			props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
			props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
			consumer = new KafkaConsumer<>(props);
			consumer.subscribe(Arrays.asList(topic));
		}

		public void execute(int workerNum) {
			executors = new ThreadPoolExecutor(workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
					new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());

			while (true) {
				ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
				for (final ConsumerRecord<String, String> record : records) {
					executors.submit(new ConsumerRecordWorker(record));
				}
			}
		}

		public void shutdown() {
			if (consumer != null) {
				consumer.close();
			}
			if (executors != null) {
				executors.shutdown();
			}
			try {
				if (!executors.awaitTermination(10, TimeUnit.SECONDS)) {
					System.out.println("Timeout.... Ignore for this case");
				}
			} catch (InterruptedException ignored) {
				System.out.println("Other thread interrupted this shutdown, ignore for this case.");
				Thread.currentThread().interrupt();
			}
		}

	}

	// 记录处理
	public static class ConsumerRecordWorker implements Runnable {

		private ConsumerRecord<String, String> record;

		public ConsumerRecordWorker(ConsumerRecord<String, String> record) {
			this.record = record;
		}

		@Override
		public void run() {
			// 假如说数据入库操作
			System.out.println("Thread - " + Thread.currentThread().getName());
			System.err.printf("patition = %d , offset = %d, key = %s, value = %s%n", record.partition(),
					record.offset(), record.key(), record.value());
		}
	}
}